/*
 * Copyright (c) 2010 Pentaho Corporation.  All rights reserved. 
 * This software was developed by Pentaho Corporation and is provided under the terms 
 * of the GNU Lesser General Public License, Version 2.1. You may not use 
 * this file except in compliance with the license. If you need a copy of the license, 
 * please go to http://www.gnu.org/licenses/lgpl-2.1.txt. The Original Code is Pentaho 
 * Data Integration.  The Initial Developer is Pentaho Corporation.
 *
 * Software distributed under the GNU Lesser Public License is distributed on an "AS IS" 
 * basis, WITHOUT WARRANTY OF ANY KIND, either express or  implied. Please refer to 
 * the license for the specific language governing your rights and limitations.
 */
package org.pentaho.di.www;

import java.util.ArrayList;
import java.util.List;
import java.util.Timer;
import java.util.TimerTask;
import java.util.UUID;
import java.util.concurrent.atomic.AtomicBoolean;

import org.pentaho.di.cluster.SlaveServer;
import org.pentaho.di.core.Const;
import org.pentaho.di.core.KettleEnvironment;
import org.pentaho.di.core.exception.KettleException;
import org.pentaho.di.core.logging.CentralLogStore;
import org.pentaho.di.core.logging.LogChannel;
import org.pentaho.di.core.logging.LogChannelInterface;
import org.pentaho.di.core.logging.LogLevel;
import org.pentaho.di.core.logging.LoggingObjectType;
import org.pentaho.di.core.logging.SimpleLoggingObject;
import org.pentaho.di.i18n.BaseMessages;
import org.pentaho.di.job.Job;
import org.pentaho.di.trans.Trans;
import org.pentaho.di.trans.TransConfiguration;
import org.pentaho.di.trans.TransExecutionConfiguration;

public class CarteSingleton {

  private static Class<?> PKG = Carte.class; // for i18n purposes, needed by Translator2!! $NON-NLS-1$

  private static SlaveServerConfig slaveServerConfig;
  private static CarteSingleton carte;

  private LogChannelInterface log;
  private TransformationMap transformationMap;
  private JobMap jobMap;
  private List<SlaveServerDetection> detections;
  private SocketRepository socketRepository;

  private CarteSingleton(SlaveServerConfig config) throws KettleException {
    KettleEnvironment.init();
    CentralLogStore.init(config.getMaxLogLines(), config.getMaxLogTimeoutMinutes());

    this.log = new LogChannel("Carte");
    transformationMap = new TransformationMap();
    transformationMap.setSlaveServerConfig(config);
    jobMap = new JobMap();
    jobMap.setSlaveServerConfig(config);
    detections = new ArrayList<SlaveServerDetection>();
    socketRepository = new SocketRepository(log);

    if (config.getObjectTimeoutMinutes() > 0) {
      installPurgeTimer(config, log, transformationMap, jobMap);
    }

    SlaveServer slaveServer = config.getSlaveServer();
    if (slaveServer != null) {
      int port = WebServer.PORT;
      if (!Const.isEmpty(slaveServer.getPort())) {
        try {
          port = Integer.parseInt(slaveServer.getPort());
        } catch (Exception e) {
          log.logError(BaseMessages.getString(PKG, "Carte.Error.CanNotPartPort", slaveServer.getHostname(), "" + port), e);
        }
      }

      // TODO: see if we need to keep doing this on a periodic basis.
      // The master might be dead or not alive yet at the time we send this
      // message.
      // Repeating the registration over and over every few minutes might
      // harden this sort of problems.
      //
      if (config.isReportingToMasters()) {
        String hostname = slaveServer.getHostname();
        final SlaveServer client = new SlaveServer("Dynamic slave [" + hostname + ":" + port + "]", hostname, "" + port, slaveServer.getUsername(), slaveServer
            .getPassword());
        for (final SlaveServer master : config.getMasters()) {
          // Here we use the username/password specified in the slave
          // server section of the configuration.
          // This doesn't have to be the same pair as the one used on the
          // master!
          //
          try {
            SlaveServerDetection slaveServerDetection = new SlaveServerDetection(client);
            master.sendXML(slaveServerDetection.getXML(), RegisterSlaveServlet.CONTEXT_PATH + "/");
            log.logBasic("Registered this slave server to master slave server [" + master.toString() + "] on address [" + master.getServerAndPort() + "]");
          } catch (Exception e) {
            log.logError("Unable to register to master slave server [" + master.toString() + "] on address [" + master.getServerAndPort() + "]");
          }
        }
      }
    }
  }

  public static void installPurgeTimer(final SlaveServerConfig config, final LogChannelInterface log, final TransformationMap transformationMap,
      final JobMap jobMap) {
    // If we need to time out finished or idle objects, we should create a timer in the background to clean
    //
    if (config.getObjectTimeoutMinutes() > 0) {

      log.logBasic("Installing timer to purge stale objects after " + config.getObjectTimeoutMinutes() + " minutes.");

      Timer timer = new Timer(true);

      final AtomicBoolean busy = new AtomicBoolean(false);
      TimerTask timerTask = new TimerTask() {
        public void run() {
          if (!busy.get()) {
            busy.set(true);

            try {
              // Check all transformations...
              //
              for (CarteObjectEntry entry : transformationMap.getTransformationObjects()) {
                Trans trans = transformationMap.getTransformation(entry);

                // See if the transformation is finished or stopped.
                //
                if (trans != null && (trans.isFinished() || trans.isStopped()) && trans.getLogDate() != null) {
                  // check the last log time
                  //
                  int diffInMinutes = (int) Math.floor((System.currentTimeMillis() - trans.getLogDate().getTime()) / 60000);
                  if (diffInMinutes >= config.getObjectTimeoutMinutes()) {
                    // Let's remove this from the transformation map...
                    //
                    transformationMap.removeTransformation(entry);
                    transformationMap.deallocateServerSocketPorts(entry);
                  }
                }
              }

              // And the jobs...
              //
              for (CarteObjectEntry entry : jobMap.getJobObjects()) {
                Job job = jobMap.getJob(entry);

                // See if the job is finished or stopped.
                //
                if (job != null && (job.isFinished() || job.isStopped()) && job.getLogDate() != null) {
                  // check the last log time
                  //
                  int diffInMinutes = (int) Math.floor((System.currentTimeMillis() - job.getLogDate().getTime()) / 60000);
                  if (diffInMinutes >= config.getObjectTimeoutMinutes()) {
                    // Let's remove this from the job map...
                    //
                    jobMap.removeJob(entry);
                  }
                }
              }

            } finally {
              busy.set(false);
            }
          }
        }
      };

      // Search for stale objects every minute:
      //
      timer.schedule(timerTask, 60000, 60000);
    }
  }

  public static CarteSingleton getInstance() {
    try {
      if (carte == null) {
        if (slaveServerConfig == null) {
          slaveServerConfig = new SlaveServerConfig();
          SlaveServer slaveServer = new SlaveServer();
          slaveServerConfig.setSlaveServer(slaveServer);
        }

        carte = new CarteSingleton(slaveServerConfig);

        Trans trans = Carte.generateTestTransformation();

        String carteObjectId = UUID.randomUUID().toString();
        SimpleLoggingObject servletLoggingObject = new SimpleLoggingObject("CarteSingleton", LoggingObjectType.CARTE, null);
        servletLoggingObject.setContainerObjectId(carteObjectId);
        servletLoggingObject.setLogLevel(LogLevel.BASIC);

        carte.getTransformationMap().addTransformation(trans.getName(), carteObjectId, trans,
            new TransConfiguration(trans.getTransMeta(), new TransExecutionConfiguration()));

        return carte;
      } else {
        return carte;
      }
    } catch (KettleException ke) {
      throw new RuntimeException(ke);
    }
  }

  public TransformationMap getTransformationMap() {
    return transformationMap;
  }

  public void setTransformationMap(TransformationMap transformationMap) {
    this.transformationMap = transformationMap;
  }

  public JobMap getJobMap() {
    return jobMap;
  }

  public void setJobMap(JobMap jobMap) {
    this.jobMap = jobMap;
  }

  public List<SlaveServerDetection> getDetections() {
    return detections;
  }

  public void setDetections(List<SlaveServerDetection> detections) {
    this.detections = detections;
  }

  public SocketRepository getSocketRepository() {
    return socketRepository;
  }

  public void setSocketRepository(SocketRepository socketRepository) {
    this.socketRepository = socketRepository;
  }

  public static SlaveServerConfig getSlaveServerConfig() {
    return slaveServerConfig;
  }

  public static void setSlaveServerConfig(SlaveServerConfig slaveServerConfig) {
    CarteSingleton.slaveServerConfig = slaveServerConfig;
  }

}
